package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author WEIYUNHUI
 * @date 2023/6/10 11:15
 * <p>
 * 算子链: 将满足条件的算子链接到一起，形成一个大的task ，该操作称之为算子链合并。
 * <p>
 * 满足的条件:
 * 上下游的并行度一致， 且数据的分发规则为forward。
 * <p>
 * 数据的分发规则(分区规则)：
 * RescalePartitioner： 相对负载均衡。
 * RebalancePartitioner：  绝对负载均衡 上游的数据会轮询发送到下游的算子
 * 如果上下游的并行度不一致， 默认就是Rebalance
 * KeyGroupStreamPartitioner：  hash ,按照key的hash对下游并行度取余。
 * GlobalPartitioner： 强制并行度为1，数据全部发往下游的第一个并行度中。
 * ForwardPartitioner： 直连， 要求上下游的并行度必须一致。
 * ShufflePartitioner： 洗牌, 随机发送下游的并行度。
 * BroadcastPartitioner：广播， 将数据广播到下游的每个并行度。
 * CustomPartitioner：
 * <p>
 * <p>
 * 禁用算子链:
 * 1. 针对于某个算子
 * startNewChain() : 从当前算子开始一个新链（不和上游算子合并，和下游算子合并）
 * disableChaining(): 当前算子禁用算子链（不与上游和下游的算子合并）
 * 2. 针对于全局
 * env.disableOperatorChaining();
 * <p>
 * 算子链的好处:
 * 它减少线程间切换、缓冲的开销，并且减少延迟的同时增加整体吞吐量。
 */
public class Flink09_OperatorChain {
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 5678);

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //设置并行度
        env.setParallelism(1);

        //全局禁用算子链
        env.disableOperatorChaining();

        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);

        ds      //.rebalance()
                //.rescale()
                //.keyBy( x -> x )
                //.shuffle()
                //.global()
                //.broadcast()
                //.forward()
                .map(x -> x).name("map1").setParallelism(1)
                .map(x -> x).name("map2").setParallelism(1)
                //.startNewChain()
                //.disableChaining()
                .print().setParallelism(1);

        // 5. 启动执行
        env.execute();
    }

}